package com.haozhen.homework

/**
  * @author haozhen
  * @email haozh@ync1.com
  * @date 2021/1/3  16:08
  */
object Work3 {


  def getAdid(url:String): String ={
    val params = url.split("&")
    val param = params(2).split("=")
    param(1)
  }

  def main(args: Array[String]): Unit = {
    import org.apache.spark.{SparkConf, SparkContext}

    val conf = new SparkConf().setAppName("work3").setMaster("local[*]")

    val context = new SparkContext(conf)


    val clickLog = context.textFile("data/click.log")

    val clickDataRDD = clickLog.map(line=>{
      val fields = line.split("\\s+")
      (getAdid(fields(3)),1)
    }).reduceByKey(_+_)

    val impLog = context.textFile("data/imp.log")

    val impDataRDD = impLog.map(line=>{
      val fields = line.split("\\s+")
      (getAdid(fields(3)),1)
    }).reduceByKey(_+_)
    val clickImpData = clickDataRDD.fullOuterJoin(impDataRDD)

    val result = clickImpData.map{
      case (key,(clickNum,impNum)) => (key,clickNum.getOrElse(0),impNum.getOrElse(0))
    }
    result.saveAsTextFile("data/clickImp")

    context.stop()
  }
}
